package com.arch.flink.window.watermark;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 *
 * 功能描述:
 * 操作步骤:
 *
 *
 * @author pizhihui
 * @date 2024-05-29 16:09
 */
public class OutofOrderDemo {


    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);


        env.addSource(new SourceFunction<Tuple2<String, Long>>() {
                    @Override
                    public void run(SourceContext<Tuple2<String, Long>> ctx) throws Exception {
                        ctx.collect(Tuple2.of("key", 0L));
                        ctx.collect(Tuple2.of("key", 1000L));
                        ctx.collect(Tuple2.of("key", 2000L));
                        ctx.collect(Tuple2.of("key", 3000L));
                        ctx.collect(Tuple2.of("key", 3000L));
                        ctx.collect(Tuple2.of("key", 4000L));
                        ctx.collect(Tuple2.of("key", 5000L));
                        // out of order
                        ctx.collect(Tuple2.of("key", 4000L));
                        ctx.collect(Tuple2.of("key", 6000L));
                        ctx.collect(Tuple2.of("key", 6000L));
                        ctx.collect(Tuple2.of("key", 7000L));
                        ctx.collect(Tuple2.of("key", 8000L));
                        ctx.collect(Tuple2.of("key", 10000L));
                        // out of order
                        ctx.collect(Tuple2.of("key", 8000L));
                        ctx.collect(Tuple2.of("key", 9000L));
                    }

                    @Override
                    public void cancel() {

                    }
                }).assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<Tuple2<String, Long>>() {
//                    private int outOfOrder = 0;
            /*
            Result
            (key, 13000)   0+1000+2000+3000+3000+4000
            (key, 32000)   5000+6000+6000+7000+8000
            (key, 10000)   10000
             */

                    private int outOfOrder = 3000;
            /*
             Result
            (key, 17000)
            (key, 49000)
            (key, 10000)
             */

                    @Nullable
                    @Override
                    public Watermark checkAndGetNextWatermark(Tuple2<String, Long> lastElement, long extractedTimestamp) {

                        long res = lastElement.f1 - outOfOrder;
                        System.out.println("每条数据,data:"+ lastElement.f1 +  " 的water: " + res);
                        return new Watermark(res);
                    }

                    @Override
                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                        return element.f1;
                    }
                }).keyBy(f -> f.f0)
                .window(TumblingEventTimeWindows.of(Time.of(5, TimeUnit.SECONDS)))
                // .allowedLateness(Time.minutes(1))
                // .sideOutputLateData()
                //.sum(1)
                .process(new ProcessWindowFunction<Tuple2<String, Long>, Tuple2<String, Long>, String, TimeWindow>() {

                    @Override
                    public void process(String key,
                                        ProcessWindowFunction<Tuple2<String, Long>, Tuple2<String, Long>, String, TimeWindow>.Context context,
                                        Iterable<Tuple2<String, Long>> elements,
                                        Collector<Tuple2<String, Long>> out) throws Exception {

                        TimeWindow window = context.window();
                        long start = window.getStart();
                        long end = window.getEnd();
                        long l = context.currentWatermark();


                        Long sum = 0L;
                        List<Long> data = new ArrayList<>();
                        for (Tuple2<String, Long> element : elements) {
                            sum += element.f1;
                            data.add(element.f1);
                        }
                        out.collect(Tuple2.of(key, sum));

                        System.err.println("now:" + new Timestamp(System.currentTimeMillis()).toString() + "[" +start+ "," + end +"),watermark=" + l + ",elements=" + Arrays.toString(data.toArray()));



                    }
                })
                .print();


        env.execute("out of order demo");


    }

}
